The poll fn needs to interact with the notified() future for this to work. Check out this for inspiration:

You should be able to do something similar without creating a whole future struct if you use tokio::pin! on the notified() future.

If we now notify only once, when the last receiver drops, can we just await the Notified directly? I originally had this block because I notified on every Rx drop, so had to include an additional check to ensure that we had just dropped the last Rx.

I'm also trying to get a better understanding of the internals here. For my own understanding - this doesn't work because, after the first call, the internal Future state machine is progressed to the point where we're already in the poll_fn and have a Pending from a non-terminal Rx dropping. That means that we'd never await the notified() again. Is that correct?

Yes, Notified doesn't have spurious wakeups in the scenario we are using it, so it should work. Well, except for the fact that we probably want it to return immediately if it's already closed when we call it. We can do it like this:

loop {
    let notified = self.shared.notify_rx_drop.notified();
    if self.is_closed() { return; }
    notified.await;
}

Sure, done. I used tail.closed instead of self.is_closed() as I couldn't find an is_closed on Sender, which means an extra lock.

For my own curiosity, what's the advantage of the loop above vs. a simple self.shared.notify_rx_drop.notified().await, if Notified doesn't suffer from spurious wakeups? Now that we only notify once when the last receiver drops, it may be a nice way to avoid that extra lock I mentioned.

If you don't use a loop, then it changes the behavior of the test I requested in #6685 (comment). Which behavior we want is up for discussion.